package com.doit.flume;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.text.SimpleDateFormat;
import java.util.Date;
import java.util.List;

/**
 * @Date: 23.6.2
 * @Author: Hang.Nian.YY
 * @qq: 598196583
 * @Tips: 学大数据 ,到多易教育
 * @Description:
 *  自定义拦截器 :
 *  抽取事件中的时间戳 , 转换成对应的日期  , 将日期放在事件对象的头中
 *  数据sink的时候  可以从头信息中取出数据对应的日期作为数据输出的文件夹
 */
public class ExtractDateInterceptor implements Interceptor {

    public  ExtractDateInterceptor(){

    }

    String key ;
    public  ExtractDateInterceptor(String key){
        this.key = key ;
    }
    SimpleDateFormat simpleDateFormat ;
    /**
     * 在拦截事件之前执行一次
     * 一般做一些初始化的工作
     */
    @Override
    public void initialize() {
        simpleDateFormat = new SimpleDateFormat("yyyy-MM-dd");
    }

    /**
     * 拦截每条数据(日志的封装 Event对象)
     * @param event
     * @return
     */
    @Override
    public Event intercept(Event event) {
        try {
            // 获取日志内容  json串 {...."timestamp":1685674444360}}
            String line = new String(event.getBody());
            // 抽取时间戳
            JSONObject jsonObject = JSON.parseObject(line);
            Long ts = jsonObject.getLong("timestamp");
            // 时间戳 -- 日期字符串
            String dt = simpleDateFormat.format(new Date(ts));
            // 放在Header中
            event.getHeaders().put(key , dt) ;
            // 返回
        } catch (Exception e) {
            event.getHeaders().put(key , "error") ;
        }
        return event;
    }

    /**
     * 拦截一批次
     * @param list
     * @return
     */
    @Override
    public List<Event> intercept(List<Event> list) {
        for (Event event : list) {
            intercept(event) ;
        }
        return list;
    }

    /**
     * 最后调用一次
     */
    @Override
    public void close() {
        // 垃圾回收机制  立马回收
        simpleDateFormat = null ;
    }
    /**
     * 拦截器的调用方式
     * 构建者模式  builder
     * 构建器   构建拦截器
     * com.doit.flume.ExtractDateInterceptor.ExtractDateInterceptorBuilder
     */
    public  static  class  ExtractDateInterceptorBuilder implements  Builder{
        String datePath ;
        @Override
        public Interceptor build() {
            return new  ExtractDateInterceptor(datePath);
        }

        /**
         * 可以用户用户的配置参数
         * 在调用build方法之前执行
         * 可以获取用户的配置信息
         * @param context
         */
        @Override
        public void configure(Context context) {
             datePath = context.getString("datePath"); //ngg
        }
    }

}
